package com.shujia.flink

import org.apache.flink.api.common.typeinfo.{TypeInformation, Types}
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.scala.BatchTableEnvironment
import org.apache.flink.table.api.scala._
import org.apache.flink.api.scala._
import org.apache.flink.core.fs.FileSystem.WriteMode
import org.apache.flink.table.sinks.CsvTableSink
import org.apache.flink.table.sources.CsvTableSource
import org.apache.flink.types.Row

object Ddemo6TableApi {
  def main(args: Array[String]): Unit = {

    val batchEnv = ExecutionEnvironment.getExecutionEnvironment

    val dataSet = batchEnv.fromCollection(List[(String, Int, String)](
      ("张三", 35, "男"),
      ("李四", 32, "男"),
      ("王五", 34, "女"),
      ("赵六", 32, "男")
    ))


    //创建table上下文对象
    val tableEnv = BatchTableEnvironment.create(batchEnv)


    //将dataSet转换成table   table:  有列名和列类型
    val table = tableEnv.fromDataSet(dataSet, 'name, 'age, 'gender)


    //产看标信息
    table.printSchema()

    table.select("name").toDataSet[Row].print()

    table.select('name, 'age.-(5)).toDataSet[Row].print()

    table.where('gender === "男").toDataSet[Row].print()

    table.filter('gender === "女").toDataSet[Row].print()

    table
      .groupBy('gender)
      .select('gender, 'age.max.as('max))
      .select('gender, 'max)
      .toDataSet[Row].print()


    println("*" * 100)

    val ds = table.toDataSet[Row]
    ds.map(row => {
      val name = row.getField(0)
      val age = row.getField(1)
      val gender = row.getField(2)
      s"$name\t$age\t$gender"
    }).print()


    //将table注册成一张表
    tableEnv.registerTable("person", table)

    tableEnv.sqlQuery(
      """
        |select gender,max(age) from person group by gender
        |
      """.stripMargin).toDataSet[Row].print()


    //构建一个输出表
    val sink = new CsvTableSink("data/flink/csv", ",", 1, WriteMode.OVERWRITE)
    val fieldNames: Array[String] = Array("gender", "max")
    val fieldTypes: Array[TypeInformation[_]] = Array(Types.STRING, Types.INT)
    tableEnv.registerTableSink("sink", fieldNames, fieldTypes, sink)

    tableEnv.sqlUpdate(
      """
        |insert into sink select gender,max(age) from person group by gender
        |
      """.stripMargin)


    // table.insertInto("sink")


    //构建tablesource
    val names: Array[String] = Array("id", "name", "age", "gender", "clazz")
    val types: Array[TypeInformation[_]] = Array(Types.STRING, Types.STRING, Types.INT, Types.STRING, Types.STRING)
    val tableSource = new CsvTableSource("data/students.txt", names, types, ",", "\n")

    //注册表
    tableEnv.registerTableSource("student", tableSource)


    //构建输出表
    val sink1 = new CsvTableSink("data/flink/clazz", ",", 1, WriteMode.OVERWRITE)
    val fieldNames1: Array[String] = Array("clazz", "c")
    val fieldTypes1: Array[TypeInformation[_]] = Array(Types.STRING, Types.LONG)
    tableEnv.registerTableSink("clazz", fieldNames1, fieldTypes1, sink1)

    val resultTable = tableEnv.sqlQuery(
      """
        |select clazz,count(1) as c from student group by clazz
        |
      """.stripMargin)

    //将数写入表
    resultTable.insertInto("clazz")


    batchEnv.execute()

  }
}
